#ifndef __M_CHANNEL_H__
#define __M_CHANNEL_H__
#include <iostream>
#include <string>
#include <unordered_map>
#include <memory>
#include <vector>
#include <mutex>
#include <sstream>
#include <unordered_map>
#include <condition_variable>

#include "../mqcommon/mq_msg.pb.h"
#include "../mqcommon/mq_proto.pb.h"
#include "../mqcommon/mq_loger.hpp"
#include "../mqcommon/mq_helper.hpp"

#include "protobuf/codec/codec.h"
#include "protobuf/codec/dispatcher.h"
#include "mq_consumer.hpp"



namespace mq{
typedef std::shared_ptr<google::protobuf::Message> MessagePtr; 
using basicCommonResponsePtr = std::shared_ptr<basicCommonResponse>;
using basicConsumeResponsePtr = std::shared_ptr<basicConsumeResponse>;
using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;

class Channel{
private:
    std::mutex _mutex;
    std::condition_variable _cv;//条件变量
    std::string _cid;//信道id

    Consumer::ptr _consumer;//信道对应的消费者
    
    muduo::net::TcpConnectionPtr _conn;//连接句柄
    ProtobufCodecPtr _codec;//协议处理句柄
    std::unordered_map<std::string,basicCommonResponsePtr> _basic_resp;//存储 返回的响应， rid ： 响应句柄
public:
    Channel(const ProtobufCodecPtr& codec,
            const muduo::net::TcpConnectionPtr& conn):
            _cid(mqhelper::UUIDHelper::uuid()),
            _conn(conn),_codec(codec){}

    ~Channel(){
        basicCancel();
    }

    std::string cid(){
        return _cid;
    }

//打开信道
    bool openChannel(){
        //1. 创建并构造请求id对象cid
        std::string rid = mqhelper::UUIDHelper::uuid();
        //创建请求
        openChannelRequest req;
        req.set_cid(_cid);
        req.set_rid(rid);

        _codec->send(_conn,req);
        basicCommonResponsePtr resp = waitResponse(rid);
        return resp->ok();

    }

//关闭信道
    void closeChannel() {
        std::string rid = mqhelper::UUIDHelper::uuid();
        closeChannelRequest req; 
        req.set_rid(rid);
        req.set_cid(_cid);
        _codec->send(_conn, req);
        waitResponse(rid);
        return ;
    }

//此处的参数必须和提起指定好的请求格式对应
//打开交换机
    bool declareExchange(const std::string& ename,
                            mq::ExchangeType etype,  
                            bool edurable,bool eauto_delete,
                            google::protobuf::Map<std::string,std::string>& eargs){
        //1. 创建并构造请求id对象cid
        std::string rid = mqhelper::UUIDHelper::uuid();
        //2. 创建请求对象，并构造其内部成员
        declareExchangeRequest req;
        req.set_rid(rid);
        req.set_cid(_cid);
        req.set_exchange_name(ename);
        req.set_exchange_type(etype);
        req.set_auto_delete(eauto_delete);
        req.set_durable(edurable);
        req.mutable_args()->swap(eargs);

        //3. 向服务器发送请求
        _codec->send(_conn,req);
        //4. 等待服务器返回响应
        basicCommonResponsePtr resp = waitResponse(rid);
        return resp->ok();//返回结果
    }

//删除交换机
    void deleteExchange(const std::string& ename){
        //1. 创建并构造请求id对象cid
        std::string rid = mqhelper::UUIDHelper::uuid();
        //2. 创建请求对象，并构造其内部成员
        deleteExchangeRequest req;
        req.set_rid(rid);
        req.set_cid(_cid);
        req.set_exchange_name(ename);

        //3. 向服务器发送请求
        _codec->send(_conn,req);
        //4. 等待服务器返回响应
        waitResponse(rid);
        return;
    }

//打开队列
    bool declareQueue(const std::string& qname,
            bool qdurable,
            bool qexclusive,
            bool qauto_delete,
            google::protobuf::Map<std::string,std::string>& qargs){
                //1. 创建并构造请求id对象cid
                std::string rid = mqhelper::UUIDHelper::uuid();
                //2. 创建请求对象，并构造其内部成员
                declareQueueRequest req;
                req.set_rid(rid);
                req.set_cid(_cid);
                req.set_queue_name(qname);
                req.set_exclusive(qexclusive);
                req.set_auto_delete(qauto_delete);
                req.set_durable(qdurable);
                req.mutable_args()->swap(qargs);

                //3. 向服务器发送请求
                _codec->send(_conn,req);
                //4. 等待服务器返回响应
                basicCommonResponsePtr resp = waitResponse(rid);
                return resp->ok();//返回结果
            }

//删除队列
        void deleteQueue(const std::string& qname){
            //1. 创建并构造请求id对象cid
            std::string rid = mqhelper::UUIDHelper::uuid();
            //2. 创建请求对象，并构造其内部成员
            deleteQueueRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(qname);

            //3. 向服务器发送请求
            _codec->send(_conn,req);
            //4. 等待服务器返回响应
            waitResponse(rid);
            return;
        }

//绑定交换机和队列 
        bool queueBind(const std::string& ename,
                const std::string& qname,
                const std::string& key){

            //1. 创建并构造请求id对象cid
            std::string rid = mqhelper::UUIDHelper::uuid();
            //2. 创建请求对象，并构造其内部成员
            queueBindRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(ename);
            req.set_queue_name(qname);
            req.set_binding_key(key);

            //3. 向服务器发送请求
            _codec->send(_conn,req);
            //4. 等待服务器返回响应
            basicCommonResponsePtr resp = waitResponse(rid);
            return resp->ok();//返回结果
        }

//解绑交换机和队列 
        void queueUnBind(const std::string& ename,
                        const std::string& qname){
            //1. 创建并构造请求id对象cid
            std::string rid = mqhelper::UUIDHelper::uuid();
            //2. 创建请求对象，并构造其内部成员
            queueUnBindRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(ename);
            req.set_queue_name(qname);

            //3. 向服务器发送请求
            _codec->send(_conn,req);
            //4. 等待服务器返回响应
            waitResponse(rid);
            return;//返回结果
        }

//发布消息
        void basicPublish(const std::string& ename,
                    BasicProperties* bp,
                    const std::string& body){
                    
            //1. 创建并构造请求id对象cid
            std::string rid = mqhelper::UUIDHelper::uuid();
            //2. 创建请求对象，并构造其内部成员
            basicPublishRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_body(body);
            req.set_exchange_name(ename);
            if(bp != nullptr){
                //如果不为空才填充数据
                req.mutable_properties()->set_id(bp->id());
                req.mutable_properties()->set_delivery_mode(bp->delivery_mode());
                req.mutable_properties()->set_routing_key(bp->routing_key());
            }
            //3. 向服务器发送请求
            _codec->send(_conn,req);
            //4. 等待服务器返回响应
            waitResponse(rid);
            return ;//返回结果

        }


//取消订阅
        void basicCancel(){
            // 0. 判断消费者是否为空，若为空则无取消对象
            if(_consumer.get() == nullptr){
                return ;//则直接返回
            }
            //1. 创建并构造请求id对象cid
            std::string rid = mqhelper::UUIDHelper::uuid();
            //2. 创建请求对象，并构造其内部成员
            basicCancelRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_consumer_tag(_consumer->tag);
            req.set_queue_name(_consumer->qname);

            //3. 向服务器发送请求
            _codec->send(_conn,req);
            //4. 等待服务器返回响应
            waitResponse(rid);

            _consumer.reset();//将智能指针重置清空
            // DLOG("订阅取消!");
            return ;//返回结果
        }

//消息确认
        void basicAck(const std::string& msg_id){
            if(_consumer.get() == nullptr){
                return ;
            }
            //1. 创建并构造请求id对象cid
            std::string rid = mqhelper::UUIDHelper::uuid();
            //2. 创建请求对象，并构造其内部成员
            basicAckRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_message_id(msg_id);
            req.set_queue_name(_consumer->qname);

            //3. 向服务器发送请求
            _codec->send(_conn,req);
            //4. 等待服务器返回响应
            waitResponse(rid);
            return ;//返回结果
        }

//订阅队列
        bool basicConsume(const std::string& tag,
                        const std::string& qname,
                        bool auto_ack,
                        const ConsumerCallback& callback){

            // 0. 判断消费者是否为空，若不为空则代表已经存在消费者了
            if(_consumer.get() != nullptr){
                DLOG("当前信道已订阅其他队列消息！");
                return false;
            }
            //1. 创建并构造请求id对象cid
            std::string rid = mqhelper::UUIDHelper::uuid();
            //2. 创建请求对象，并构造其内部成员
            basicConsumeRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_consumer_tag(tag);
            req.set_queue_name(qname);
            req.set_auto_ack(auto_ack);

            //3. 向服务器发送请求
            _codec->send(_conn,req);
            // DLOG("发送消费请求已服务器");
            //4. 等待服务器返回响应
            basicCommonResponsePtr resp = waitResponse(rid);

            // std::cout << "------------" << std::endl;

            //5. 判断消费者订阅是否成功
            if(resp->ok() == false){
                DLOG("添加订阅失败！");
                return false;
            }
            // DLOG("消费者订阅成功");
            _consumer = std::make_shared<Consumer>(tag,qname,auto_ack,callback);//注意shared_ptr只是个类型！！不能用来构建
            
            return true;//返回结果
        }
public:
    //连接收到基础响应后向hash中添加 
    void putBasicResponse(const basicCommonResponsePtr& resp){
        //1. 加锁
        std::unique_lock<std::mutex> _lock(_mutex);
        //2. 将响应插入到map中
        _basic_resp.insert(std::make_pair(resp->rid(),resp));
        //3. 唤醒所有条件变量
        _cv.notify_all();
    }

//连接收到消息推送后，需要通过信道找到消费者对象，通过回调函数进行消息处理
    void consume(const basicConsumeResponsePtr& resp){
        if(_consumer.get() == nullptr){
            DLOG("消息处理时，未找到订阅者信息！");
            return ;//则直接返回空
        }
        //判断返回回来的消费者的标识，和当前信道消费者的标识是否一致
        if(resp->consumer_tag() != _consumer->tag){
            DLOG("收到的推送消息中的消费者标识，与当前信道消费者标识不一致！") ;return;
        }
        //调用回调函数
        _consumer->callback(resp->consumer_tag(),resp->mutable_properties(),resp->body());
    }

    using ptr = std::shared_ptr<Channel>;

private:
    basicCommonResponsePtr waitResponse(const std::string& rid){
        //1. 加锁
        std::unique_lock<std::mutex> _lock(_mutex);
        // DLOG("等待响应...")
        //2. 调用条件变量等待同步
        _cv.wait(_lock,[&rid,this](){
            return _basic_resp.find(rid) != _basic_resp.end();//当存在响应时返回
        });

        // DLOG("获得一个响应！");
        //3. 获取该响应
        basicCommonResponsePtr resp = _basic_resp[rid];
        //4. 将响应从map中删除，因为已经获取到了
        _basic_resp.erase(rid);
        return resp;
    }

};
class ChannelManager{
private:
    std::mutex _mutex;
    std::unordered_map<std::string,Channel::ptr> _channels;//管理信道的map
public:
    using ptr = std::shared_ptr<ChannelManager>;


    Channel::ptr create(const ProtobufCodecPtr& codec,
            const muduo::net::TcpConnectionPtr& conn){
        //1.加锁
        std::unique_lock<std::mutex> _lock(_mutex);
    
        //创建
        Channel::ptr channel = std::make_shared<Channel>(codec,conn);
        //插入map中
        _channels.insert(std::make_pair(channel->cid(),channel));
        DLOG("信道 %s 创建成功",channel->cid().c_str());
        return channel;
    }

    void remove(const std::string& cid){
        //1.加锁
        std::unique_lock<std::mutex> _lock(_mutex);
        DLOG("信道 %s 删除成功",cid.c_str());
        //删除
        _channels.erase(cid);
    }

    Channel::ptr get(const std::string& cid){
        //1.加锁
        std::unique_lock<std::mutex> _lock(_mutex);
        auto c = _channels.find(cid);
        if(c == _channels.end()){
            DLOG("%s信道未找到",cid.c_str());
            return Channel::ptr();//若不存在则直接返回空
        }
        return c->second;
    }
};


}
#endif